# -*- coding: utf-8 -*-
"""
Python 企业级Lambda函数最佳实践示例

本模块演示了Python Lambda函数的企业级最佳实践，包括:
- 基础Lambda函数和高阶函数
- 函数式编程模式
- 装饰器和闭包
- 偏函数和柯里化
- 函数组合和管道
- 错误处理和类型安全
- 性能优化和缓存
- 异步Lambda和并发处理
"""

import asyncio
import functools
import itertools
import logging
import operator
import time
from collections import defaultdict, namedtuple
from dataclasses import dataclass
from enum import Enum, auto
from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union, Iterator
from concurrent.futures import ThreadPoolExecutor, as_completed
import weakref

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(levelname)s:%(name)s:%(message)s'
)
logger = logging.getLogger(__name__)

# 类型变量
T = TypeVar('T')
U = TypeVar('U')
V = TypeVar('V')

class OperationType(Enum):
    """操作类型枚举"""
    ARITHMETIC = auto()
    COMPARISON = auto()
    LOGICAL = auto()
    STRING = auto()
    COLLECTION = auto()
    ASYNC = auto()

@dataclass
class FunctionMetrics:
    """函数性能指标"""
    name: str
    call_count: int = 0
    total_time: float = 0.0
    avg_time: float = 0.0
    min_time: float = float('inf')
    max_time: float = 0.0
    error_count: int = 0

class FunctionProfiler:
    """函数性能分析器"""
    
    def __init__(self):
        self._metrics: Dict[str, FunctionMetrics] = defaultdict(
            lambda: FunctionMetrics(name="unknown")
        )
    
    def profile(self, func_name: str = None):
        """装饰器：分析函数性能"""
        def decorator(func):
            name = func_name or func.__name__
            
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.perf_counter()
                try:
                    result = func(*args, **kwargs)
                    return result
                except Exception as e:
                    self._metrics[name].error_count += 1
                    raise
                finally:
                    end_time = time.perf_counter()
                    duration = end_time - start_time
                    
                    metrics = self._metrics[name]
                    metrics.name = name
                    metrics.call_count += 1
                    metrics.total_time += duration
                    metrics.avg_time = metrics.total_time / metrics.call_count
                    metrics.min_time = min(metrics.min_time, duration)
                    metrics.max_time = max(metrics.max_time, duration)
            
            return wrapper
        return decorator
    
    def get_metrics(self, func_name: str) -> Optional[FunctionMetrics]:
        """获取函数性能指标"""
        return self._metrics.get(func_name)
    
    def get_all_metrics(self) -> Dict[str, FunctionMetrics]:
        """获取所有性能指标"""
        return dict(self._metrics)
    
    def reset_metrics(self) -> None:
        """重置性能指标"""
        self._metrics.clear()

# 全局性能分析器
profiler = FunctionProfiler()

class LambdaFactory:
    """Lambda函数工厂类"""
    
    @staticmethod
    def create_arithmetic_ops() -> Dict[str, Callable[[float, float], float]]:
        """创建算术运算Lambda函数"""
        return {
            'add': lambda x, y: x + y,
            'subtract': lambda x, y: x - y,
            'multiply': lambda x, y: x * y,
            'divide': lambda x, y: x / y if y != 0 else float('inf'),
            'power': lambda x, y: x ** y,
            'modulo': lambda x, y: x % y if y != 0 else 0,
            'floor_div': lambda x, y: x // y if y != 0 else 0,
        }
    
    @staticmethod
    def create_comparison_ops() -> Dict[str, Callable[[Any, Any], bool]]:
        """创建比较运算Lambda函数"""
        return {
            'equal': lambda x, y: x == y,
            'not_equal': lambda x, y: x != y,
            'less_than': lambda x, y: x < y,
            'less_equal': lambda x, y: x <= y,
            'greater_than': lambda x, y: x > y,
            'greater_equal': lambda x, y: x >= y,
            'is_same': lambda x, y: x is y,
            'is_not_same': lambda x, y: x is not y,
        }
    
    @staticmethod
    def create_string_ops() -> Dict[str, Callable]:
        """创建字符串操作Lambda函数"""
        return {
            'to_upper': lambda s: s.upper() if isinstance(s, str) else str(s).upper(),
            'to_lower': lambda s: s.lower() if isinstance(s, str) else str(s).lower(),
            'capitalize': lambda s: s.capitalize() if isinstance(s, str) else str(s).capitalize(),
            'reverse': lambda s: s[::-1] if isinstance(s, str) else str(s)[::-1],
            'length': lambda s: len(s) if hasattr(s, '__len__') else 0,
            'strip': lambda s: s.strip() if isinstance(s, str) else str(s).strip(),
            'split': lambda s, sep=' ': s.split(sep) if isinstance(s, str) else str(s).split(sep),
            'join': lambda items, sep='': sep.join(str(item) for item in items),
        }
    
    @staticmethod
    def create_collection_ops() -> Dict[str, Callable]:
        """创建集合操作Lambda函数"""
        return {
            'filter_even': lambda items: list(filter(lambda x: x % 2 == 0, items)),
            'filter_odd': lambda items: list(filter(lambda x: x % 2 != 0, items)),
            'map_square': lambda items: list(map(lambda x: x ** 2, items)),
            'map_double': lambda items: list(map(lambda x: x * 2, items)),
            'reduce_sum': lambda items: functools.reduce(lambda x, y: x + y, items, 0),
            'reduce_product': lambda items: functools.reduce(lambda x, y: x * y, items, 1),
            'reduce_max': lambda items: functools.reduce(lambda x, y: max(x, y), items) if items else None,
            'reduce_min': lambda items: functools.reduce(lambda x, y: min(x, y), items) if items else None,
            'group_by': lambda items, key_func: {k: list(g) for k, g in itertools.groupby(sorted(items, key=key_func), key_func)},
            'partition': lambda items, predicate: (list(filter(predicate, items)), list(filter(lambda x: not predicate(x), items))),
        }

class FunctionalPipeline:
    """函数式编程管道"""
    
    def __init__(self, initial_value: Any = None):
        self.value = initial_value
        self.operations: List[Tuple[str, Callable]] = []
    
    def pipe(self, func: Callable, name: str = None) -> 'FunctionalPipeline':
        """添加管道操作"""
        operation_name = name or getattr(func, '__name__', 'anonymous')
        self.operations.append((operation_name, func))
        return self
    
    def execute(self, initial_value: Any = None) -> Any:
        """执行管道操作"""
        current_value = initial_value if initial_value is not None else self.value
        
        for operation_name, func in self.operations:
            try:
                current_value = func(current_value)
                logger.debug(f"管道操作 '{operation_name}' 完成，结果: {current_value}")
            except Exception as e:
                logger.error(f"管道操作 '{operation_name}' 失败: {e}")
                raise
        
        return current_value
    
    def reset(self) -> 'FunctionalPipeline':
        """重置管道"""
        self.operations.clear()
        return self
    
    def clone(self) -> 'FunctionalPipeline':
        """克隆管道"""
        new_pipeline = FunctionalPipeline(self.value)
        new_pipeline.operations = self.operations.copy()
        return new_pipeline

def curry(func: Callable) -> Callable:
    """柯里化装饰器"""
    @functools.wraps(func)
    def curried(*args, **kwargs):
        if len(args) + len(kwargs) >= func.__code__.co_argcount:
            return func(*args, **kwargs)
        return lambda *more_args, **more_kwargs: curried(*(args + more_args), **{**kwargs, **more_kwargs})
    return curried

def compose(*functions: Callable) -> Callable:
    """函数组合"""
    def composed(value):
        for func in reversed(functions):
            value = func(value)
        return value
    return composed

def memoize(maxsize: int = 128) -> Callable:
    """记忆化装饰器"""
    def decorator(func):
        cache = {}
        cache_order = []
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 创建缓存键
            key = str(args) + str(sorted(kwargs.items()))
            
            if key in cache:
                return cache[key]
            
            result = func(*args, **kwargs)
            
            # 管理缓存大小
            if len(cache) >= maxsize:
                oldest_key = cache_order.pop(0)
                del cache[oldest_key]
            
            cache[key] = result
            cache_order.append(key)
            
            return result
        
        wrapper.cache_info = lambda: {'size': len(cache), 'maxsize': maxsize}
        wrapper.cache_clear = lambda: (cache.clear(), cache_order.clear())
        
        return wrapper
    return decorator

def safe_lambda(default_value: Any = None, log_errors: bool = True) -> Callable:
    """安全Lambda装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if log_errors:
                    logger.warning(f"Lambda函数执行失败: {e}")
                return default_value
        return wrapper
    return decorator

async def async_map(func: Callable, items: List[Any], max_workers: int = 5) -> List[Any]:
    """异步映射函数"""
    async def async_func(item):
        if asyncio.iscoroutinefunction(func):
            return await func(item)
        else:
            # 在线程池中运行同步函数
            loop = asyncio.get_event_loop()
            return await loop.run_in_executor(None, func, item)
    
    # 限制并发数量
    semaphore = asyncio.Semaphore(max_workers)
    
    async def bounded_async_func(item):
        async with semaphore:
            return await async_func(item)
    
    tasks = [bounded_async_func(item) for item in items]
    return await asyncio.gather(*tasks)

def demonstrate_basic_lambdas() -> None:
    """演示基础Lambda函数"""
    print("\n======== 1. 基础Lambda函数 ========")
    
    # 算术运算
    arithmetic_ops = LambdaFactory.create_arithmetic_ops()
    
    print("✓ 算术运算:")
    test_values = [(10, 3), (15, 4), (8, 2)]
    for x, y in test_values:
        results = {name: op(x, y) for name, op in arithmetic_ops.items()}
        print(f"  {x}, {y}: {results}")
    
    # 比较运算
    comparison_ops = LambdaFactory.create_comparison_ops()
    
    print("\n✓ 比较运算:")
    test_pairs = [(5, 5), (3, 7), (10, 2)]
    for x, y in test_pairs:
        results = {name: op(x, y) for name, op in comparison_ops.items()}
        print(f"  {x} vs {y}: {results}")

def demonstrate_string_operations() -> None:
    """演示字符串操作Lambda函数"""
    print("\n======== 2. 字符串操作Lambda函数 ========")
    
    string_ops = LambdaFactory.create_string_ops()
    test_strings = ["  Hello World  ", "python programming", "Lambda Functions"]
    
    for test_str in test_strings:
        print(f"\n✓ 原始字符串: '{test_str}'")
        for name, op in string_ops.items():
            try:
                if name == 'split':
                    result = op(test_str)
                elif name == 'join':
                    result = op(['a', 'b', 'c'], '-')
                else:
                    result = op(test_str)
                print(f"  {name}: {result}")
            except Exception as e:
                print(f"  {name}: 错误 - {e}")

def demonstrate_collection_operations() -> None:
    """演示集合操作Lambda函数"""
    print("\n======== 3. 集合操作Lambda函数 ========")
    
    collection_ops = LambdaFactory.create_collection_ops()
    test_data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    
    print(f"✓ 测试数据: {test_data}")
    
    for name, op in collection_ops.items():
        try:
            if name == 'group_by':
                result = op(test_data, lambda x: x % 3)
            elif name == 'partition':
                result = op(test_data, lambda x: x % 2 == 0)
            else:
                result = op(test_data)
            print(f"  {name}: {result}")
        except Exception as e:
            print(f"  {name}: 错误 - {e}")

def demonstrate_functional_pipeline() -> None:
    """演示函数式编程管道"""
    print("\n======== 4. 函数式编程管道 ========")
    
    # 数据处理管道
    data = list(range(1, 11))
    print(f"✓ 原始数据: {data}")
    
    pipeline = FunctionalPipeline(data)
    result = (pipeline
              .pipe(lambda x: list(filter(lambda n: n % 2 == 0, x)), "过滤偶数")
              .pipe(lambda x: list(map(lambda n: n ** 2, x)), "平方")
              .pipe(lambda x: list(map(lambda n: n * 2, x)), "乘以2")
              .pipe(lambda x: sum(x), "求和")
              .execute())
    
    print(f"✓ 管道处理结果: {result}")
    
    # 字符串处理管道
    text = "  Hello, Functional Programming!  "
    print(f"\n✓ 原始文本: '{text}'")
    
    text_pipeline = FunctionalPipeline()
    text_result = (text_pipeline
                   .pipe(lambda s: s.strip(), "去除空格")
                   .pipe(lambda s: s.lower(), "转小写")
                   .pipe(lambda s: s.replace(',', ''), "移除逗号")
                   .pipe(lambda s: s.split(), "分割单词")
                   .pipe(lambda words: [word for word in words if len(word) > 3], "过滤短词")
                   .execute(text))
    
    print(f"✓ 文本处理结果: {text_result}")

def demonstrate_currying_and_composition() -> None:
    """演示柯里化和函数组合"""
    print("\n======== 5. 柯里化和函数组合 ========")
    
    # 柯里化示例
    @curry
    def multiply_three(x, y, z):
        return x * y * z
    
    # 部分应用
    multiply_by_2 = multiply_three(2)
    multiply_by_2_and_3 = multiply_by_2(3)
    
    print("✓ 柯里化示例:")
    print(f"  multiply_three(2, 3, 4) = {multiply_three(2, 3, 4)}")
    print(f"  multiply_by_2(3, 4) = {multiply_by_2(3, 4)}")
    print(f"  multiply_by_2_and_3(4) = {multiply_by_2_and_3(4)}")
    
    # 函数组合示例
    add_one = lambda x: x + 1
    multiply_by_two = lambda x: x * 2
    square = lambda x: x ** 2
    
    composed_func = compose(square, multiply_by_two, add_one)
    
    print("\n✓ 函数组合示例:")
    test_values = [1, 2, 3, 4, 5]
    for value in test_values:
        result = composed_func(value)
        step_by_step = square(multiply_by_two(add_one(value)))
        print(f"  compose(square, *2, +1)({value}) = {result} (验证: {step_by_step})")

def demonstrate_memoization() -> None:
    """演示记忆化"""
    print("\n======== 6. 记忆化和缓存 ========")
    
    @memoize(maxsize=10)
    @profiler.profile("fibonacci")
    def fibonacci(n):
        if n <= 1:
            return n
        return fibonacci(n - 1) + fibonacci(n - 2)
    
    # 测试斐波那契数列
    print("✓ 斐波那契数列计算:")
    test_numbers = [10, 15, 20, 10, 15]  # 重复计算测试缓存
    
    for n in test_numbers:
        start_time = time.perf_counter()
        result = fibonacci(n)
        end_time = time.perf_counter()
        
        print(f"  fibonacci({n}) = {result}, 耗时: {end_time - start_time:.6f}秒")
    
    # 显示缓存信息
    cache_info = fibonacci.cache_info()
    print(f"\n✓ 缓存信息: {cache_info}")
    
    # 显示性能指标
    metrics = profiler.get_metrics("fibonacci")
    if metrics:
        print(f"✓ 性能指标: 调用{metrics.call_count}次, 平均耗时{metrics.avg_time:.6f}秒")

def demonstrate_safe_lambdas() -> None:
    """演示安全Lambda函数"""
    print("\n======== 7. 安全Lambda函数 ========")
    
    # 安全除法
    safe_divide = safe_lambda(default_value=0)(lambda x, y: x / y)
    
    print("✓ 安全除法测试:")
    test_cases = [(10, 2), (15, 3), (8, 0), ("a", "b")]
    
    for x, y in test_cases:
        result = safe_divide(x, y)
        print(f"  safe_divide({x}, {y}) = {result}")
    
    # 安全类型转换
    safe_int = safe_lambda(default_value=0)(lambda x: int(x))
    safe_float = safe_lambda(default_value=0.0)(lambda x: float(x))
    
    print("\n✓ 安全类型转换:")
    test_values = ["123", "45.67", "invalid", None, []]
    
    for value in test_values:
        int_result = safe_int(value)
        float_result = safe_float(value)
        print(f"  '{value}' -> int: {int_result}, float: {float_result}")

async def demonstrate_async_lambdas() -> None:
    """演示异步Lambda函数"""
    print("\n======== 8. 异步Lambda函数 ========")
    
    # 异步处理函数
    async def async_square(x):
        await asyncio.sleep(0.01)  # 模拟异步操作
        return x ** 2
    
    # 同步处理函数
    def sync_cube(x):
        time.sleep(0.01)  # 模拟耗时操作
        return x ** 3
    
    test_data = list(range(1, 11))
    print(f"✓ 测试数据: {test_data}")
    
    # 异步映射
    start_time = time.perf_counter()
    async_results = await async_map(async_square, test_data, max_workers=3)
    async_time = time.perf_counter() - start_time
    
    print(f"✓ 异步平方结果: {async_results}")
    print(f"✓ 异步处理耗时: {async_time:.4f}秒")
    
    # 同步映射（用于对比）
    start_time = time.perf_counter()
    sync_results = await async_map(sync_cube, test_data[:5], max_workers=3)  # 减少数据量
    sync_time = time.perf_counter() - start_time
    
    print(f"✓ 同步立方结果: {sync_results}")
    print(f"✓ 同步处理耗时: {sync_time:.4f}秒")

async def run_async_tests() -> None:
    """运行异步测试"""
    await demonstrate_async_lambdas()

def run_comprehensive_tests() -> None:
    """运行综合测试套件"""
    print("\n======== 9. 综合测试套件 ========")
    
    test_functions = [
        ("基础Lambda函数测试", demonstrate_basic_lambdas),
        ("字符串操作测试", demonstrate_string_operations),
        ("集合操作测试", demonstrate_collection_operations),
        ("函数式管道测试", demonstrate_functional_pipeline),
        ("柯里化和组合测试", demonstrate_currying_and_composition),
        ("记忆化测试", demonstrate_memoization),
        ("安全Lambda测试", demonstrate_safe_lambdas),
    ]
    
    passed = 0
    total = len(test_functions)
    
    for test_name, test_func in test_functions:
        try:
            test_func()
            print(f"✓ {test_name} 通过")
            passed += 1
        except Exception as e:
            print(f"✗ {test_name} 失败: {e}")
            logger.exception(f"测试失败: {test_name}")
    
    # 运行异步测试
    try:
        asyncio.run(run_async_tests())
        print(f"✓ 异步Lambda函数测试 通过")
        passed += 1
        total += 1
    except Exception as e:
        print(f"✗ 异步Lambda函数测试 失败: {e}")
        logger.exception("异步测试失败")
        total += 1
    
    # 显示所有性能指标
    all_metrics = profiler.get_all_metrics()
    if all_metrics:
        print("\n✓ 性能指标汇总:")
        for name, metrics in all_metrics.items():
            print(f"  {name}: {metrics.call_count}次调用, 平均{metrics.avg_time:.6f}秒")
    
    print(f"\n测试结果: {passed}/{total} 通过")
    print("所有企业级Lambda函数最佳实践演示完成!")

if __name__ == "__main__":
    run_comprehensive_tests()